package spring.cloud.tasks.executor.name_space.executor.task;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.common.node.BlockNodePaths;
import spring.cloud.tasks.common.node.TaskNodePath;
import spring.cloud.tasks.executor.name_space.NameSpaceConfiguration;
import spring.cloud.tasks.executor.name_space.NameSpaceDataManager;
import spring.cloud.tasks.executor.name_space.NameSpaceDataManagerImpl;
import spring.cloud.tasks.executor.name_space.TaskExecutorNodeCacheOrTreeCacheManager;
import spring.cloud.tasks.executor.name_space.executor.ExecutorIdToTaskIdToTaskExecutorMapMap;
import spring.cloud.tasks.executor.name_space.executor.TaskInitService;
import spring.cloud.tasks.executor.name_space.executor.task.task_item.TaskItemExecuteInfoService;
import spring.cloud.tasks.executor.name_space.executor.task.task_listener.TaskListenerHolderManager;
import spring.cloud.tasks.executor.name_space.executor.task.task_listener.task_configuration.TaskConfiguration;
import spring.cloud.tasks.executor.name_space.executor.task.task_listener.task_configuration.TaskConfigurationService;
import spring.cloud.tasks.executor.name_space.executor.task.task_listener.trigger_task_execute.TaskExecutorInfoService;
import spring.cloud.tasks.executor.name_space.executor.tasks_runtime_info.task_item_assign.TaskIdToTaskInstanceIdToTaskItemIndexListMapMap;
import spring.cloud.tasks.executor.utils.LogUtils;

import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 接受是否该执行器是否允许运行任务,如果允许则到了时间执行。执行完毕后。如果该任务没有分配到上面则自己释放自己。如果允许任务跑在上面则创建任务执行器。
 */
@Slf4j
@Setter
@Getter
public class TaskExecutor {
    private final NameSpaceDataManager nameSpaceDataManager;

    private final TaskExecutorNodeCacheOrTreeCacheManager taskExecutorNodeCacheOrTreeCacheManager;
    //
    private final TaskConfiguration taskConfiguration;
    private final TaskConfigurationService taskConfigurationService;
    private final TaskListenerHolderManager taskListenerHolderManager;
    private final TaskExecutorInfoService taskExecutorInfoService;
    private final TaskItemExecuteInfoService taskItemExecuteInfoService;
    public volatile boolean needStop = false;//外部尝试关闭任务执行器
    public ReentrantLock lock = new ReentrantLock();
    private String executorId;
    private TaskInitService taskInitService;
    private String taskId;
    private TaskConfiguration previousTaskConfiguration = new TaskConfiguration(null, null);
    private TaskExecuteEngine taskExecuteEngine;
    private AtomicBoolean shutdown = new AtomicBoolean(false);
    private ScheduledExecutorService taskExecuteInfoCollectScheduledExecutorService;

    public TaskExecutor(final NameSpaceDataManager nameSpaceDataManager, TaskInitService taskInitService, final TaskConfiguration taskConfiguration) {
        this.nameSpaceDataManager = nameSpaceDataManager;
        this.executorId = nameSpaceDataManager.getExecutorId();
        this.taskInitService = taskInitService;
        this.taskId = taskConfiguration.getTaskId();
        this.taskConfiguration = taskConfiguration;

        this.taskExecutorNodeCacheOrTreeCacheManager = new TaskExecutorNodeCacheOrTreeCacheManager(nameSpaceDataManager.getCuratorFramework(), executorId, taskId);
        //
        this.previousTaskConfiguration.setCronExpression(taskConfiguration.getCronExpression());
        this.previousTaskConfiguration.setPauseDatePeriod(taskConfiguration.getPauseDatePeriod());
        this.previousTaskConfiguration.setPauseTimePeriod(taskConfiguration.getPauseTimePeriod());
        //
        this.taskConfigurationService = new TaskConfigurationService(this);

        this.taskExecutorInfoService = new TaskExecutorInfoService(this);
        this.taskItemExecuteInfoService = new TaskItemExecuteInfoService(this);
        //

        this.taskListenerHolderManager = new TaskListenerHolderManager(this);
        ExecutorIdToTaskIdToTaskExecutorMapMap.addTaskExecutor(executorId, taskId, this);
    }

    public void needStop() {
        this.needStop = true;
        if (taskExecuteEngine != null) {
            taskExecuteEngine.needStop = true;
        }
    }

    /**
     * 初始化任务.
     */
    public void init() {
        try {
            startAll();
            createTaskExecuteEngine();
            taskExecutorInfoService.persistExecutorOnline();
        } catch (Throwable t) {
            shutdown();
            throw t;
        }
    }

    private void startAll() {
        taskListenerHolderManager.start();
        //
        taskConfigurationService.start();

        taskExecutorInfoService.start();
        taskItemExecuteInfoService.start();
        //
        taskExecutorInfoService.clearRunOneTimePath();
        taskExecutorInfoService.clearStopOneTimePath();
        taskExecutorInfoService.resetCount();


    }

    private void createTaskExecuteEngine() {
        taskExecuteEngine = new TaskExecuteEngine();
        taskExecuteEngine.setNameSpace(nameSpaceDataManager.getNameSpace());
        taskExecuteEngine.setExecutorId(executorId);
        taskExecuteEngine.setTaskId(taskId);
        taskExecuteEngine.setTaskExecutor(this);
        //
        taskExecuteEngine.setTaskConfigurationService(taskConfigurationService);
        taskExecuteEngine.setTaskItemExecuteInfoService(taskItemExecuteInfoService);
        //

        initTaskExecuteInfoCollectScheduledExecutorService();
    }

    /**
     * 立刻启动任务.
     */
    public void rightNowExecute() {

        //taskExecuteEngine.getClicker().rightNowExecute();
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                taskExecuteEngine.startTaskItemExecuteThread();
            }
        });
        thread.start();
    }

    public void shutdown() {
        synchronized (shutdown) {
            shutdown.set(true);

            // 关闭Listener
            taskListenerHolderManager.stop();

            // 关闭任务：关闭调度器，强杀业务
            stopTaskExecuteInfoCollectScheduledExecutorService();

            // 关闭服务

            taskConfigurationService.stop();

            taskExecutorInfoService.stop();


            // 关闭TreeCache
            taskExecutorNodeCacheOrTreeCacheManager.shutdown();
            deleteTaskExecutorNode();

            // 移除任务注册表
            ExecutorIdToTaskIdToTaskExecutorMapMap.removeTaskExecutor(executorId, taskId);
            taskInitService.removeTaskId(taskId);
        }
    }

    /**
     * 删除ZK结点
     */
    public void deleteTaskExecutorNode() {
        NameSpaceConfiguration nameSpaceConfiguration = ((NameSpaceDataManagerImpl) nameSpaceDataManager).getNameSpaceConfiguration();
        NameSpaceDataManagerImpl nameSpaceDataManager = new NameSpaceDataManagerImpl(nameSpaceConfiguration);
        try {
            nameSpaceDataManager.init(); // maybe throw RuntimeException
            nameSpaceDataManager.delete(BlockNodePaths.getExecutorTaskNode(executorId, taskId));//任务分配
            nameSpaceDataManager.delete(TaskNodePath.getTaskExecutorNode(taskId, executorId));//只是标记
        } catch (Throwable t) {
            LogUtils.error(log, taskId, "delete job node error", t);
        } finally {
            nameSpaceDataManager.close();
        }
    }

    /**
     * 调度器分配的任务
     */
    public TaskIdToTaskInstanceIdToTaskItemIndexListMapMap getTaskIdToTaskInstanceIdToTaskItemIndexListMapMap() {
        TaskIdToTaskInstanceIdToTaskItemIndexListMapMap taskIdToTaskInstanceIdToTaskItemIndexListMapMap = taskInitService.getTaskIdToTaskInstanceIdToTaskItemIndexListMapMap();
        return taskIdToTaskInstanceIdToTaskItemIndexListMapMap;
    }

    public ConcurrentHashMap<String, List<Integer>> getTaskInstanceIdToItemIndexListMap() {
        TaskIdToTaskInstanceIdToTaskItemIndexListMapMap taskIdToTaskInstanceIdToTaskItemIndexListMapMap = taskInitService.getTaskIdToTaskInstanceIdToTaskItemIndexListMapMap();
        return taskIdToTaskInstanceIdToTaskItemIndexListMapMap.getTaskInstanceIdToItemIndexListMap(taskId);
    }

    public void initTaskExecuteInfoCollectScheduledExecutorService() {
        if (taskExecuteInfoCollectScheduledExecutorService != null) {
            taskExecuteInfoCollectScheduledExecutorService.shutdown();
        }
        taskExecuteInfoCollectScheduledExecutorService = getTaskExecuteInfoCollectScheduledExecutorService(this);
        TaskStateMaintenanceThread taskStateMaintenanceThread = new TaskStateMaintenanceThread(this);
        taskExecuteInfoCollectScheduledExecutorService.scheduleAtFixedRate(taskStateMaintenanceThread, 10, 10, TimeUnit.SECONDS);
    }

    public void stopTaskExecuteInfoCollectScheduledExecutorService() {
        if (taskExecuteInfoCollectScheduledExecutorService != null) {
            taskExecuteInfoCollectScheduledExecutorService.shutdown();
        }
    }

    public ScheduledExecutorService getTaskExecuteInfoCollectScheduledExecutorService(final TaskExecutor taskExecutor) {
        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {//
            @Override
            public Thread newThread(Runnable runnable) {
                Thread thread = new Thread(runnable, taskExecutor.getExecutorId() + "_" + taskExecutor.getTaskConfigurationService().getTaskId() + TaskStateMaintenanceThread.class.getSimpleName());
                thread.setDaemon(false);
                thread.setPriority(Thread.NORM_PRIORITY);
                return thread;
            }
        });
        return scheduledExecutorService;
    }
}
